package storm2;// -*- codeing: utf-8 -*-

// @Time     :2021/12/8 22:22
// @Author   :郭荣杰
//@File      :KafkaSourceSpout.java
//@Software  : IntelliJ IDEA
import kafka.api.OffsetRequest;
import org.apache.storm.kafka.BrokerHosts;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.ZkHosts;

/**
 * *spout1 接上话题，得到order表数据
 */

public class KafkaSourceSpout_order {

    public KafkaSpout createKafkSpout(){
        /**
         * 描述zookeeper端口
         * 描述kafka话题名
         * 根据zookeeper动态连接kafka话题进行消费数据
         */
        String brokerZkStr = "192.168.179.141:2181,192.168.179.142:2181,192.168.179.143:2181";
        BrokerHosts hosts = new ZkHosts(brokerZkStr);   // 通过zookeeper中的/brokers即可找到kafka的地址
        String topic = "order";
        String zkRoot = "/" + topic;
        String id = "consumer-id";
        SpoutConfig spoutConf = new SpoutConfig(hosts, topic, zkRoot, id);
        // 本地环境设置之后，也可以在zk中建立/f-k-s节点，在集群环境中，不用配置也可以在zk中建立/f-k-s节点
        //spoutConf.zkServers = Arrays.asList(new String[]{"uplooking01", "uplooking02", "uplooking03"});
        //spoutConf.zkPort = 2181;
        spoutConf.startOffsetTime = OffsetRequest.LatestTime(); // 设置之后，刚启动时就不会把之前的消费也进行读取，会从最新的偏移量开始读取
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConf);
        return kafkaSpout;
    }

}
